Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

Run ID: ebda7b1b-2998-4f72-a46e-4207104ed513

Reviewing files that changed from the base of the PR and between 5ee5959 and cfe56ab.

• references/ai-chat/src/app/actions.ts is excluded by !references/**
• references/ai-chat/src/components/chat-app.tsx is excluded by !references/**
• references/ai-chat/src/components/chat.tsx is excluded by !references/**
• references/ai-chat/src/trigger/chat.ts is excluded by !references/**

• docs/guides/ai-chat.mdx
• packages/core/src/v3/realtimeStreams/manager.ts
• packages/core/src/v3/realtimeStreams/noopManager.ts
• packages/core/src/v3/realtimeStreams/streamInstance.ts
• packages/core/src/v3/realtimeStreams/streamsWriterV1.ts
• packages/core/src/v3/realtimeStreams/streamsWriterV2.ts
• packages/core/src/v3/realtimeStreams/types.ts
• packages/trigger-sdk/src/v3/ai.ts
• packages/trigger-sdk/src/v3/chat.ts
• packages/trigger-sdk/src/v3/streams.ts

• GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
• GitHub Check: sdk-compat / Node.js 22.12 (ubuntu-latest)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 8)
• GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 8)
• GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - npm)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
• GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 8)
• GitHub Check: units / packages / 🧪 Unit Tests: Packages (1, 1)
• GitHub Check: sdk-compat / Cloudflare Workers
• GitHub Check: sdk-compat / Deno Runtime
• GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
• GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
• GitHub Check: typecheck / typecheck
• GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
• GitHub Check: sdk-compat / Node.js 20.20 (ubuntu-latest)
• GitHub Check: sdk-compat / Bun Runtime
• GitHub Check: Analyze (javascript-typescript)
• GitHub Check: Analyze (python)

packages/trigger-sdk/src/v3/chat.ts (4)

278-279: Pass headers option to sendInputStream API call.

The extraHeaders stored on the transport are used for SSE subscriptions (Line 415) but not passed to the sendInputStream call here. This breaks proxy/tenant routing or custom auth flows that depend on these headers.

Proposed fix

         const apiClient = new ApiClient(this.baseURL, session.publicAccessToken);
-        await apiClient.sendInputStream(session.runId, CHAT_MESSAGES_STREAM_ID, minimalPayload);
+        await apiClient.sendInputStream(session.runId, CHAT_MESSAGES_STREAM_ID, minimalPayload, {
+          headers: this.extraHeaders,
+        });

---

297-302: Pass headers option to triggerTask API call.

Same issue as sendInputStream — the extraHeaders are not forwarded to the triggerTask call, breaking custom auth or proxy routing.

Proposed fix

-    const triggerResponse = await apiClient.triggerTask(this.taskId, {
-      payload,
-      options: {
-        payloadType: "application/json",
-      },
-    });
+    const triggerResponse = await apiClient.triggerTask(
+      this.taskId,
+      {
+        payload,
+        options: {
+          payloadType: "application/json",
+        },
+      },
+      undefined,
+      { headers: this.extraHeaders }
+    );

---

438-441: Pass headers option to stop signal sendInputStream call.

The stop signal path also omits extraHeaders.

Proposed fix

             const api = new ApiClient(this.baseURL, session.publicAccessToken);
             api
-              .sendInputStream(session.runId, CHAT_STOP_STREAM_ID, { stop: true })
+              .sendInputStream(session.runId, CHAT_STOP_STREAM_ID, { stop: true }, {
+                headers: this.extraHeaders,
+              })
               .catch(() => {}); // Best-effort

---

459-550: LGTM — Stream subscription and chunk processing logic.

The SSE stream subscription correctly:

• Handles __trigger_turn_complete control chunks and token refresh
• Tracks lastEventId for stream resumption
• Manages abort signals and graceful shutdown
• Skips leftover chunks after stop via skipToTurnComplete

packages/trigger-sdk/src/v3/ai.ts (6)

247-248: Global _chatPipeCount is race-prone with concurrent chat runs.

_chatPipeCount is module-global, so concurrent chat runs can mutate the same counter and incorrectly suppress or trigger auto-piping in other runs.

Suggested fix — scope counter per run

-let _chatPipeCount = 0;
+const _chatPipeCountByRun = new Map<string, number>();

 async function pipeChat(
   source: UIMessageStreamable | AsyncIterable<unknown> | ReadableStream<unknown>,
   options?: PipeChatOptions
 ): Promise<void> {
-  _chatPipeCount++;
+  const runId = taskContext.ctx?.run.id;
+  if (runId) {
+    _chatPipeCountByRun.set(runId, (_chatPipeCountByRun.get(runId) ?? 0) + 1);
+  }
   // ...
 }

Then in the chatTask turn loop, reset and check the run-scoped counter instead of the global.

---

453-552: chatTask does not support custom streamKey end-to-end.

TriggerChatTransport allows configuring a custom streamKey, but chatTask hardcodes CHAT_STREAM_KEY for auto-piping (Line 801) and the turn-complete control chunk (Line 1047). If a user configures a non-default stream key on the frontend, the turn completion will hang because the frontend subscribes to a different stream than where the backend writes.

Suggested fix — add `streamKey` option to `chatTask`

Add a streamKey option to ChatTaskOptions and thread it through to pipeChat and writeTurnCompleteChunk:

 export type ChatTaskOptions<TIdentifier extends string> = Omit<...> & {
+  /** Stream key for output. Must match TriggerChatTransport.streamKey. `@default` "chat" */
+  streamKey?: string;
   run: (payload: ChatTaskRunPayload) => Promise<unknown>;
   // ...
 };

 function chatTask<TIdentifier extends string>(...) {
   const {
     run: userRun,
+    streamKey = CHAT_STREAM_KEY,
     // ...
   } = options;

   // In auto-pipe:
-  await pipeChat(uiStream, { signal: combinedSignal, spanName: "stream response" });
+  await pipeChat(uiStream, { streamKey, signal: combinedSignal, spanName: "stream response" });

   // In turn complete:
-  const turnCompleteResult = await writeTurnCompleteChunk(...);
+  const turnCompleteResult = await writeTurnCompleteChunk(streamKey, ...);
 }

-async function writeTurnCompleteChunk(chatId?: string, publicAccessToken?: string) {
-  const { waitUntilComplete } = streams.writer(CHAT_STREAM_KEY, {
+async function writeTurnCompleteChunk(streamKey: string = CHAT_STREAM_KEY, chatId?: string, publicAccessToken?: string) {
+  const { waitUntilComplete } = streams.writer(streamKey, {

---

880-884: Verify handling of multiple buffered messages.

When messages arrive during streaming, they're buffered in pendingMessages. However, only the first message is used (Line 882) — any additional messages are discarded. If multiple messages arrive during a streaming turn, subsequent ones will be lost.

Consider preserving the backlog:

+      const queuedMessages: ChatTaskWirePayload[] = [];
+
       // ... inside turn loop after processing ...
       if (pendingMessages.length > 0) {
-        currentWirePayload = pendingMessages[0]!;
+        currentWirePayload = pendingMessages.shift()!;
+        queuedMessages.push(...pendingMessages);
         return "continue";
       }
+
+      // Check queued messages before waiting
+      if (queuedMessages.length > 0) {
+        currentWirePayload = queuedMessages.shift()!;
+        return "continue";
+      }

The past review indicated this was addressed. Please verify the intended behavior — is dropping extra messages intentional (e.g., expecting only one message per turn)?

---

734-777: LGTM — Lifecycle hooks implementation.

The onChatStart and onTurnStart hooks are correctly sequenced and wrapped in traced spans. The scoped access token is minted per-turn and passed to callbacks for persistence.

---

691-713: LGTM — Message accumulation logic.

The accumulation correctly handles:

• Turn 0: Full history from frontend initializes the accumulator
• Regenerate: Full history resets the accumulator (removing last assistant message)
• Submit (turn 1+): Only new messages appended to existing accumulator

This aligns with the frontend transport sending minimal payloads after the first turn.

---

1096-1115: LGTM — stripProviderMetadata helper.

This correctly strips ephemeral OpenAI itemId fields that would cause 404 errors when sent back in subsequent streamText calls. The implementation preserves other provider metadata while removing only the problematic fields.

docs/guides/ai-chat.mdx (2)

20-22: Verify the ai package version requirement.

The documentation states the ai package v5.0.0 or later is required. A past review indicated the reference implementation uses v6.0.0. Please verify and update the version requirement if needed.

#!/bin/bash
# Check ai package version in reference implementations and package.json files
rg -n '"ai"\s*:' --glob '**/package.json' | head -20

---

344-406: LGTM — Persistence documentation.

The persistence section clearly explains:

• What needs to be persisted (messages + sessions)
• Server-side persistence via onTurnStart (before streaming) and onTurnComplete (after)
• The purpose of lastEventId for stream resumption
• Complete code examples with database operations

packages/core/src/v3/realtimeStreams/types.ts (1)

29-31: StreamWriteResult is a clean API extension.

This shape keeps backward compatibility while enabling resume metadata.

packages/core/src/v3/realtimeStreams/manager.ts (1)

19-21: Type propagation for active stream waits is consistent.

Good alignment with the new wait(): Promise<StreamWriteResult> contract.

packages/core/src/v3/realtimeStreams/noopManager.ts (1)

18-18: Noop manager now matches the wait contract.

Returning an object here correctly aligns with StreamWriteResult.

packages/core/src/v3/realtimeStreams/streamsWriterV1.ts (1)

261-264: wait() return type update is correctly wired.

This keeps V1 aligned with the shared writer contract.

packages/core/src/v3/realtimeStreams/streamInstance.ts (1)

66-69: StreamInstance.wait() now correctly preserves writer output.

Returning writer.wait() is the right propagation point for StreamWriteResult.

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts (1)

173-176: Capturing lastSeqNum and exposing it from wait() is a solid improvement.

This makes V2 waits materially more useful for resume flows.

Also applies to: 188-191

packages/trigger-sdk/src/v3/streams.ts (1)

143-201: Span customization and waitpoint entity attribution are well integrated.

spanName/collapsed wiring and waitpointId attribution improve trace clarity.

Also applies to: 757-775, 822-823